package utils

import (
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

type GoTaskV0 struct {
	isTerminated int32
	worker_num   int32 // 当前工作纤程数量
	channel_size int32 // 组数量
	taskcnt      int32 // 当前任务数量
	popcnt       int32
	pushfailcnt  int32
	started      int32

	config_max_work_num         int32 // 最多工作纤程数(这些工作纤程, 轮询执行任务)
	channel_max_queue_size      int32
	channelPool                 sync.Pool // Pool可以使用,但是用来做队列不行
	channelMap                  sync.Map
	config_channel_max_work_num sync.Map // groupid:max-work-num 设置通道最大纤程数时使用, 如果创建通道时会获取

	closeWg sync.WaitGroup

	Monitor_push *MonitorSpeedNTimeRec
}

func NewGoTaskV0() *GoTaskV0 {
	rval := &GoTaskV0{
		Monitor_push:           NewMonitorSpeedNTimeRec(),
		config_max_work_num:    128,
		channel_max_queue_size: 1024,
	}
	rval.channelPool.New = rval.onNewChannel
	return rval
}

func (this *GoTaskV0) onNewChannel() interface{} {
	val := &multi_group_task_channel{
		task_current_work_cnt: 0,
		work_max_cnt:          1,
		max_queue_size:        this.channel_max_queue_size,
		dataQueue:             NewSyncQueue(),
		state:                 0,
	}
	return val
}

func (this *GoTaskV0) onTaskFuncNoArgs(sender *MultiGroupTaskRec) {
	cbFun := sender.paramFunc.(func())
	if cbFun != nil {
		cbFun()
	}
}

func (this *GoTaskV0) cleanup(channel *multi_group_task_channel) {
	for {
		rec := channel.Pop()
		if rec == nil {
			break
		}
		atomic.AddInt32(&this.taskcnt, -1)
		atomic.AddInt32(&this.popcnt, 1)

		rec.Close()
	}
}

/**
 * 如果channel 运行多个纤程工作, 通一个channel会有多纤程同时执行
 */
func (this *GoTaskV0) doChannelWork(channel *multi_group_task_channel) {
	for {
		rec := channel.Pop()
		if rec == nil {
			break
		}
		atomic.AddInt32(&this.taskcnt, -1)
		atomic.AddInt32(&this.popcnt, 1)

		if this.isTerminated == 1 {
			rec.err = errors.New("[工作停止]请勿执行耗时操作")
			//} else if err != nil {
			//	cachelist.err = err
		}
		rec.CallBack()
	}
}

func (this *GoTaskV0) innerDoWork(ch *multi_group_task_channel) {
	if GoFunCatchException {
		defer PanicHandler()
	}
	defer ch.endWork()
	goid := GetCurrentGoRoutineID()
	this.closeWg.Add(1)
	defer func() {
		this.closeWg.Done()
		atomic.AddInt32(&this.worker_num, -1)
	}()
	ch.last_goroute_id = goid
	ch.last_exec_start_T = time.Now()
	this.doChannelWork(ch)
}

func (this *GoTaskV0) checkStart() {
	if atomic.CompareAndSwapInt32(&this.started, 0, 1) {
		go func() {
			n := 0
			this.channelMap.Range(func(key, value interface{}) bool {
				ch := value.(*multi_group_task_channel)
				if ch.tryBeginWork() {
					if atomic.AddInt32(&this.worker_num, 1) >= this.config_max_work_num { // 工作纤程不够
						atomic.AddInt32(&this.worker_num, -1)
						ch.endWork()
						return false
					}
					go this.innerDoWork(ch)
					n++
				}
				return true
			})
			if n == 0 {
				time.Sleep(100)
			} else {
				time.Sleep(0)
			}
		}()
	}
}

func (this *GoTaskV0) PostTaskFunc(groupid interface{}, cb func()) error {
	rec := &MultiGroupTaskRec{
		Args:      nil,
		paramFunc: cb,
		Cb:        this.onTaskFuncNoArgs,
	}
	return this.PostTask(groupid, rec)
}

func (this *GoTaskV0) releaseChannel(val *multi_group_task_channel) {
	this.channelPool.Put(val)
}

func (this *GoTaskV0) PostTask(groupid interface{}, rec *MultiGroupTaskRec) error {
	if this.isTerminated == 1 {
		return errors.New("[停止工作]不能压入任务")
	}
	if DEBUG_MODE == 1 {
		rec.CallStack = fmt.Sprintf("[%s]%s", NowString(), GetCallStack(1))
	}
	rec.groupID = groupid
	newChannel := this.channelPool.Get().(*multi_group_task_channel)
	loadObj, loaded := this.channelMap.LoadOrStore(groupid, newChannel)
	if loaded {
		this.releaseChannel(newChannel)
	} else { // 使用的新的
		newChannel.id = groupid
		//newChannel.task_current_work_cnt = 0   // 不能进行赋值, 避免被借出去后,直接beginWork操作了该值...
		if cnt, loaded := this.config_channel_max_work_num.Load(groupid); loaded {
			newChannel.work_max_cnt = cnt.(int32)
			if newChannel.work_max_cnt < 0 {
				newChannel.work_max_cnt = 1
			}
		}
		atomic.AddInt32(&this.channel_size, 1)
	}
	realChannel := loadObj.(*multi_group_task_channel)

	this.Monitor_push.Delta(1)
	atomic.AddInt32(&this.taskcnt, 1)
	err := realChannel.Push(rec)
	if err != nil {
		// 压入失败 -1
		atomic.AddInt32(&this.taskcnt, -1)
		atomic.AddInt32(&this.pushfailcnt, 1)
	} else {
		//if this.FixedWorker == 0 {
		//	if this.config_grow_work_if_task_num > 0 && this.taskcnt > this.config_grow_work_if_task_num { // 大于100个任务未执行
		//
		//		mg_patrol.AddTaskOnceA1(realChannel, func(arg interface{}) {
		//			ch := arg.(*multi_group_task_channel)
		//			if ch.needWorker() {
		//				this.checkGrowWork() // 增长worker
		//			}
		//		})
		//	}
		//}
	}

	return err
}
